package com.develop.alarm.service;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import com.develop.alarm.bean.AlarmInfo;
import com.develop.common.PropertiesUtils;
import com.sun.org.apache.xpath.internal.operations.And;

import net.sf.json.JSONObject;


  
public class AlarmConsumerAll {
	/**
	 * 告警数据保存，从kafka获取全部告警数据，保存入库
	 * 接收参数如下
	 * topicName:订阅消息主题名称  例如：BJ_ERICSSON_realTimeAlarm
	 * province:省份名称   例如：BJ(北京)
	 * firmName:厂家名称    例如：ERICSSON(爱立信)
	* <p>Description: </p> 
	* @author xxj
	* @date 2019年7月9日
	 */
	public static void saveAlarm(String province,String firmName,String topicName){
		 Properties props = new Properties();  
	        props.put("auto.offset.reset", "earliest"); //必须要加，如果要读旧数据
	        props.put("bootstrap.servers", PropertiesUtils.getPropertiesByKey("kafka.consumer.bootstrap.servers", "application.properties"));
	        props.put("group.id", "testgroup");  
	        props.put("enable.auto.commit", "true");  
	        props.put("auto.commit.interval.ms", "1000");  
	        props.put("session.timeout.ms", "30000");
	        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
	        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
	        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
	        TopicPartition topicPartition1=new TopicPartition(topicName,0);
	        consumer.assign(Arrays.asList(topicPartition1));
	        consumer.seek(topicPartition1,0);
	        
//	        List<TopicPartition> topicPartitions=new ArrayList<TopicPartition>();
//	        topicPartitions.add(topicPartition1);
//	        consumer.seekToBeginning(topicPartitions);
//	        consumer.subscribe(Arrays.asList(topicName));
	        LinkedList<AlarmInfo> alarmList=new LinkedList<AlarmInfo>();
	        //告警批量处理条数
	        int batchnum=Integer.parseInt(PropertiesUtils.getPropertiesByKey("batchnum", "application.properties"));
	        
	        while (true) {
	            ConsumerRecords<String, String> records = consumer.poll(1000);  
	            if(records.count()>=batchnum){
	            	System.out.println("=============="+records.count());
	            	for (ConsumerRecord<String, String> record : records){
	            		JSONObject jsonobject = JSONObject.fromObject(record.value());
	            		AlarmInfo alarm=(AlarmInfo)JSONObject.toBean(jsonobject,AlarmInfo.class);
	            		alarmList.add(alarm);
	            		if(alarmList.size()==batchnum){
	            			AlarmService.batchInsertAlarmInfo(alarmList, province, firmName);
	            			int begin=alarmList.get(0).getAlarmSeq();
	            			int end=alarmList.get(batchnum-1).getAlarmSeq();
	            			if(end-begin==batchnum-1){
	            				System.out.println("=============该批次告警数据连续======当前系统时间："+new Date()+"告警发生时间："+alarmList.get(0).getEventTime());
	            				System.out.println("开始告警序号："+alarmList.get(0).getAlarmSeq()+"结束告警序号："+alarmList.get(batchnum-1).getAlarmSeq());
	            			}else{
	            				System.out.println("=============该批次告警数据连续======当前系统时间："+new Date()+"告警发生时间："+alarmList.get(0).getEventTime());
	            				System.out.println("开始告警序号："+alarmList.get(0).getAlarmSeq()+"结束告警序号："+alarmList.get(batchnum-1).getAlarmSeq());
	            				
	            				//Todo 告知接收程序不连续告警开始序号，告警程序通知告警Socket服务端从不连续告警序号开始重新发送
	            				
	            			}
	            			alarmList.clear();
	            		}
	            	}  
	            }else if(records.count()>0&& records.count()<batchnum){
	            	int rcount=records.count();
	            	System.out.println("-------------"+rcount);
	            	for (ConsumerRecord<String, String> record : records){
	            		JSONObject jsonobject = JSONObject.fromObject(record.value());
	            		AlarmInfo alarm=(AlarmInfo)JSONObject.toBean(jsonobject,AlarmInfo.class);
	            		alarmList.add(alarm);
	            	}
	            	AlarmService.batchInsertAlarmInfo(alarmList, province, firmName);
	            	int begin=alarmList.get(0).getAlarmSeq();
	            	int end=alarmList.get(rcount-1).getAlarmSeq();
	            	if(end-begin==rcount-1){
	            		System.out.println("=============该批次告警数据连续======当前系统时间："+new Date()+"告警发生时间："+alarmList.get(0).getEventTime());
	            		System.out.println("开始告警序号："+alarmList.get(0).getAlarmSeq()+"结束告警序号："+alarmList.get(rcount-1).getAlarmSeq());
	            	}else{
	            		System.out.println("=============该批次告警数据连续======当前系统时间："+new Date()+"告警发生时间："+alarmList.get(0).getEventTime());
	            		System.out.println("开始告警序号："+alarmList.get(0).getAlarmSeq()+"结束告警序号："+alarmList.get(rcount-1).getAlarmSeq());
	            		
	            		//Todo 告知接收程序不连续告警开始序号，告警程序通知告警Socket服务端从不连续告警序号开始重新发送
	            	}
	            	alarmList.clear();
	            }
	            
	        }
	}
    public static void main(String[] args) {  
        String province=args[0];
        String firmName=args[1];
        String topicName=args[2];
        saveAlarm(province, firmName,topicName);
    }
}
